package com.atguigu0.streaming

import kafka.serializer.StringDecoder
import org.apache.kafka.clients.consumer.ConsumerConfig
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * @description: 高级api,重启后能恢复消费已经堆积的消息
 * @time: 2020/6/16 10:44
 * @author: baojinlong
 **/
object HighKafkaStreaming2 {
  val ckPath = "E:/test-data/input/kafka.ck"

  def main(args: Array[String]): Unit = {
    // 获取ssc
    val ssc: StreamingContext = StreamingContext.getActiveOrCreate(ckPath, () => createSSC())
    // 开启
    ssc.start()
    ssc.awaitTermination()
  }

  def createSSC(): StreamingContext = {
    // 创建SparkConf
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("wordCount")
    // 创建SteamingContext
    val ssc = new StreamingContext(sparkConf, Seconds(3))
    // 设置checkPoint地址
    ssc.checkpoint(ckPath)

    // 设置kafka参数
    val group: String = "group01"
    val brokers: String = "localhost:9092"
    val topic: String = "test1"
    val deserializationClass: String = "org.apache.kafka.common.serialization.StringDeserializer"

    // 封装Kafka参数
    val kafkaPara: Map[String, String] = Map[String, String](
      ConsumerConfig.GROUP_ID_CONFIG -> group,
      ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> brokers,
      ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG -> deserializationClass,
      ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG -> deserializationClass
    )

    // 读取kafka数据创建DStream
    val kafkaStream: InputDStream[(String, String)] = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaPara, Set(topic))
    // 打印数据
    kafkaStream.print
    // 返回结果
    ssc
  }

}
